package com.gitee.xhs.joining.intervalJoin;

import com.gitee.xhs.dao.IntervalJoinDao;
import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.util.Collector;

/**
 * @author BigHead
 * @date 2022/7/21 9:55
 * @description OperatorIntervalJoin
 */
public class OperatorIntervalJoin extends ProcessJoinFunction<Transaction, Transaction, IntervalJoinDao> {

    @Override
    public void processElement(Transaction left, Transaction right, ProcessJoinFunction<Transaction, Transaction, IntervalJoinDao>.Context ctx, Collector<IntervalJoinDao> out) throws Exception {
        IntervalJoinDao intervalJoinDao = new IntervalJoinDao();
        intervalJoinDao.setLeftTransaction(left);
        if (left.getTimestamp() == right.getTimestamp()) {
            intervalJoinDao.setRightEqualTransaction(right);
        }
        if (left.getTimestamp() > right.getTimestamp()) {
            intervalJoinDao.setRightLowerTransaction(right);
        }
        if (left.getTimestamp() < right.getTimestamp()) {
            intervalJoinDao.setRightUpperTransaction(right);
        }
        out.collect(intervalJoinDao);
    }

}
